#ifndef __M_CONNECTION_H__
#define __M_CONNECTION_H__
#include "muduo/proto/dispatcher.h"
#include "muduo/proto/codec.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"
#include "mq_channel.hpp"
#include "mq_worker.hpp"

namespace mymq
{
    class Connection
    {
    public:
        using ptr = std::shared_ptr<Connection>;
        Connection(const std::string& s_ip, int s_port, const Async::ptr& worker)
        : _latch(1),
          _client(worker->_loopthread.startLoop(), muduo::net::InetAddress(s_ip, s_port), "Client"),
          _dispatcher(std::bind(&Connection::onUnknownMessage, this, 
            std::placeholders::_1, std::placeholders::_2, std::placeholders::_3)),
          _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, 
            std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
          _worker(worker),
          _channel_manager(std::make_shared<ChannelManager>())
        {
            // 注册事件处理函数
            _dispatcher.registerMessageCallback<basicCommonResponse>(std::bind(&Connection::basicResponse, 
                this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeResponse>(std::bind(&Connection::consumeResponse, this, 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            
            // 告知客户端，接收到什么事件应该如何处理
            _client.setConnectionCallback(
                std::bind(&Connection::onConnection, this, std::placeholders::_1));
            _client.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), 
                std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // 连接
            _client.connect();
            // 发送连接后进行阻塞，等待连接成功后被唤醒
            _latch.wait();
        }

        // 打开信道
        Channel::ptr openChannel()
        {
            // 客户端创建信道，加入到管理
            Channel::ptr channel = _channel_manager->create(_conn, _codec);
            // 服务器也要创建与客户端对应的信道
            bool ret = channel->openChannel();
            if (ret == false)
            {
                DLOG("打开信道失败！");
                return Channel::ptr();
            }

            return channel;
        }

        // 关闭信道
        void closeChannel(const Channel::ptr& channel)
        {
            // 服务端信道关闭
            channel->closeChannel();
            // 客户端信道关闭
            _channel_manager->remove(channel->getCid());
        }
    private:
        // 接收到基本响应，插入到信道的基础响应 hash_map
        void basicResponse(const muduo::net::TcpConnectionPtr&, 
            const basicCommonResponsePtr& message, muduo::Timestamp)
        {
            // 获取信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到信道信息！");
                return;
            }
            // 将收到的响应插入到信道的基础响应hash_map中
            channel->putBasicResponse(message);
        }
        // 接收到消息，将信道消费者的回调消息处理函数进行封装，抛入线程池
        void consumeResponse(const muduo::net::TcpConnectionPtr&, 
            const basicConsumeResponsePtr message, muduo::Timestamp)
        {
            // 获取信道
            Channel::ptr channel = _channel_manager->get(message->cid());
            if (channel.get() == nullptr)
            {
                DLOG("未找到信道信息！");
                return;
            }
            // 一个信道对应一个消费者，封装消费者的消息处理回调函数，抛入线程池
            // 使用 lambda 表达式封装
            _worker->_pool.push([channel, message](){
                channel->consume(message);
            });
        }

        void onUnknownMessage(const muduo::net::TcpConnectionPtr&, 
            const MessagePtr& message, muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
        }

        void onConnection(const muduo::net::TcpConnectionPtr& conn)
        {
            if (conn->connected())
            {
                // 唤醒主线程
                _latch.countDown();
                // 连接建立成功
                _conn = conn;
            }
            else
            {
                // 连接断开
                _conn.reset();
            }
        }
    private:
        muduo::CountDownLatch _latch; // 实现同步
        muduo::net::TcpConnectionPtr _conn; // 客户端连接
        muduo::net::TcpClient _client; // 客户端
        ProtobufDispatcher _dispatcher; // 请求分发器
        ProtobufCodecPtr _codec; // 协议处理器

        Async::ptr _worker; // 异步工作模块
        ChannelManager::ptr _channel_manager; // 信道管理
    };
}

#endif